
from typing import List
from x_com import xport_core
from x_com import source_api
from x_com import topic
from datetime import datetime
import x_com_lib
import csv
import asyncio
logger = xport_core.getLogger()

class EcommerceDataSourceService:
  # 是否在推送销售数据
  is_playing_product_sale_data: bool
  # 是否在推送用户注册数据
  is_playing_user_register_data: bool
   # 是否在推送用户登录数据
  is_playing_user_login_data: bool

    
  async def on_init(self):
    # 获取数据历经
    self.data_path = x_com_lib.config.get_str("data-path")
    # 初始化为 False
    self.is_playing_product_sale_data = False
    self.is_playing_user_register_data = False
    self.is_playing_user_login_data = False
    logger.info("on_init...")

  async def on_finalize(self):
    logger.info("on_finalize...")

  def _subscribe_data(self, conn_id: int, topic:str):
      subscibe_info = x_com_lib.PySubscribeTopicInfo()
      subscibe_info.topic = topic
      subscibe_info.conn_id = conn_id
      subscibe_info.tag = []
      x_com_lib.stream.xport_subscribe_topic(subscibe_info)



  # SubscribeData
  async def subscribe_data(self, ctx: xport_core.Context):
    logger.info(f"订阅数据。。。")
    self._subscribe_data(ctx.conn_id, "PlayProductSaleData")
    self._subscribe_data(ctx.conn_id, "PlayUserRegisterData")
    self._subscribe_data(ctx.conn_id, "PlayUserLoginData")
    
  # 推送数据
  async def _play_product_sale_data(self):
    # 这里路径需改
    data_path = f"{self.data_path}/商品销售.csv"
    push_num = 0
    with open(data_path, mode='r', encoding='utf-8') as file:
       # 创建一个CSV读取器
       reader = csv.reader(file)
       # 跳过第一行（表头）
       next(reader)
       # 遍历CSV文件中的每一行
       for row in reader:
         # 
         product_sale_info = source_api.ProductSaleInfo()
         product_sale_info.user_id = row[0]
         product_sale_info.product_id = row[1]
         product_sale_info.product_category = row[2]
         # 转换为整型
         product_sale_info.num = int(row[3])
         # 转换为浮点型
         product_sale_info.price = float(row[4])
         #  
         product_sale_info.buy_time = row[5]
         topic.product_sale(product_sale_info)
         push_num += 1
         logger.info(f"已推送商品销售数据：{push_num} 条")
         # 100毫秒 推送一条数据
         await asyncio.sleep(0.1)  # 模拟处理延时
    self.is_playing_product_sale_data = True


  # PlayProductSaleData
  async def play_product_sale_data(self, ctx: xport_core.Context):
    if self.is_playing_product_sale_data:
      logger.info("商品数据在推送中...")
      return
    # 开启新的协程去推送，否则会卡住，无法返回
    asyncio.create_task(self._play_product_sale_data())

  # 推送数据
  async def _play_user_register_data(self):
    # 这里路径需改
    data_path = f"{self.data_path}/用户数据.csv"
    push_num = 0
    with open(data_path, mode='r', encoding='utf-8') as file:
       # 创建一个CSV读取器
       reader = csv.reader(file)
       # 跳过第一行（表头）
       next(reader)
       # 遍历CSV文件中的每一行
       for row in reader:
         # 
         user_info = source_api.UserInfo()
         user_info.user_id = row[0]
         if row[1] == "男":
           user_info.gender = source_api.Gender.male
         else:
            user_info.gender = source_api.Gender.female
         user_info.age = int(row[2])
         user_info.register_time = row[3]
         topic.user_register(user_info)
         push_num += 1
         logger.info(f"已推送用户注册数据：{push_num} 条")
         # 100毫秒 推送一条数据
         await asyncio.sleep(0.1)  # 模拟处理延时
    self.is_playing_user_register_data = True

  # PlayUserRegisterData
  async def play_user_register_data(self, ctx: xport_core.Context):
    if self.is_playing_user_register_data:
      logger.info("用户注册数据在推送中...")
      return
    # 开启新的协程去推送，否则会卡住，无法返回
    asyncio.create_task(self._play_user_register_data())

  # 推送数据
  async def _play_user_login_data(self):
    # 这里路径需改
    data_path = f"{self.data_path}/用户登录.csv"
    push_num = 0
    # 对于登录数据，需要将每一条数据拆分为 登录和登出数据
    login_infos = []
    with open(data_path, mode='r', encoding='utf-8') as file:
       # 创建一个CSV读取器
       reader = csv.reader(file)
       # 跳过第一行（表头）
       next(reader)
       # 遍历CSV文件中的每一行
       for row in reader:
        # 登录数据  
        login_info = source_api.UserLoginInfo()
        login_info.is_login = True
        login_info.user_id = row[0]
        login_info.handle_time = row[1]
        login_infos.append(login_info)

        # 登出数据  
        logout_info = source_api.UserLoginInfo()
        logout_info.is_login = False
        logout_info.user_id = row[0]
        logout_info.handle_time = row[2]

        login_infos.append(logout_info)
    # 根据时间从小到大排序
    sorted_login_infos:List[source_api.UserLoginInfo] = sorted(login_infos, key=lambda x: datetime.strptime(x.handle_time, '%Y-%m-%d %H:%M:%S'))
    total_num = len(sorted_login_infos)
    for login_info in sorted_login_infos:
      push_num += 1
      logger.info(f"已推送用户登录数据：{push_num} /{total_num} 条: {login_info.handle_time}")
      # 100毫秒 推送一条数据
      topic.user_login(login_info)
      await asyncio.sleep(0.1)  # 模拟处理延时
      # 根据登录等处时间

    self.is_playing_user_login_data = True

  # PlayUserLoginData
  async def play_user_login_data(self, ctx: xport_core.Context):
    if self.is_playing_user_login_data:
      logger.info("用户登录数据在推送中...")
      return
    # 开启新的协程去推送，否则会卡住，无法返回
    asyncio.create_task(self._play_user_login_data())
